package javacourse.mq.kafka.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;


/**
 * Create by chenhd on 2021/10/31.
 */
@Configuration
public class Consumer {

    @KafkaListener(topics = {"test.topic"}, containerFactory = "kafkaListenerContainerFactory")
    public void consume(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {
        try {
            System.out.println("get message event.");
            for (ConsumerRecord consumerRecord : consumerRecords) {
                System.out.println("get message:" + consumerRecord.value());
            }
        } catch (Exception e) {
            //log
            e.printStackTrace();
        } finally {
            System.out.println("submit offset.");
            ack.acknowledge();
        }
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory(ConsumerFactory consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setBatchListener(true);
        factory.getContainerProperties().setPollTimeout(1500);
        //每处理完业务手动调用Acknowledgment.acknowledge()后立即提交
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}
